package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.VisitorStats;
import com.atguigu.gmall.realtime.utils.ClickHouseUtil;
import com.atguigu.gmall.realtime.utils.DateTimeUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;

/**
 * Author: Felix
 * Date: 2022/1/4
 * Desc: 访客主题统计
 * 需要启动的进程
 * zk、kafka、logger.sh、BaseLogApp、UniqueVisitorApp、UserJumpDetailApp、VisitorStatsApp
 */
public class VisitorStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 并行度设置
        env.setParallelism(4);

        //TODO 2.检查点设置(略)

        //TODO 3.从Kafka主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String pageLogTopic = "dwd_page_log";
        String uvTopic = "dwm_unique_visitor";
        String ujdTopic = "dwm_user_jump_detail";
        String groupId = "visitor_stats_app_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> pageLogKafkaSource = MyKafkaUtil.getKafkaSource(pageLogTopic, groupId);
        FlinkKafkaConsumer<String> uvKafkaSource = MyKafkaUtil.getKafkaSource(uvTopic, groupId);
        FlinkKafkaConsumer<String> ujdKafkaSource = MyKafkaUtil.getKafkaSource(ujdTopic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> pageLogStrDS = env.addSource(pageLogKafkaSource);
        DataStreamSource<String> uvStrDS = env.addSource(uvKafkaSource);
        DataStreamSource<String> ujdStrDS = env.addSource(ujdKafkaSource);

        //pageLogStrDS.print(">>>");
        //uvStrDS.print("###");
        //ujdStrDS.print("$$$$");

        //TODO 4.对读取的流中的数据类型进行转换        jsonStr-->VisitorStats实体类对象
        //4.1 pageLog
        SingleOutputStreamOperator<VisitorStats> pageLogStatsDS = pageLogStrDS.map(
            new MapFunction<String, VisitorStats>() {
                @Override
                public VisitorStats map(String jsonStr) throws Exception {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                    JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                    VisitorStats visitorStats = new VisitorStats(
                        "",
                        "",
                        commonJsonObj.getString("vc"),
                        commonJsonObj.getString("ch"),
                        commonJsonObj.getString("ar"),
                        commonJsonObj.getString("is_new"),
                        0L,
                        1L,
                        0L,
                        0L,
                        pageJsonObj.getLong("during_time"),
                        jsonObj.getLong("ts")
                    );

                    String lastPageId = pageJsonObj.getString("last_page_id");
                    if (lastPageId == null || lastPageId.length() == 0) {
                        visitorStats.setSv_ct(1L);
                    }
                    return visitorStats;
                }
            }
        );
        //4.2 uv
        SingleOutputStreamOperator<VisitorStats> uvStatsDS = uvStrDS.map(
            new MapFunction<String, VisitorStats>() {
                @Override
                public VisitorStats map(String jsonStr) throws Exception {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                    VisitorStats visitorStats = new VisitorStats(
                        "",
                        "",
                        commonJsonObj.getString("vc"),
                        commonJsonObj.getString("ch"),
                        commonJsonObj.getString("ar"),
                        commonJsonObj.getString("is_new"),
                        1L,
                        0L,
                        0L,
                        0L,
                        0L,
                        jsonObj.getLong("ts")
                    );
                    return visitorStats;
                }
            }
        );
        //4.3 ujd
        SingleOutputStreamOperator<VisitorStats> ujdStatsDS = ujdStrDS.map(
            new MapFunction<String, VisitorStats>() {
                @Override
                public VisitorStats map(String jsonStr) throws Exception {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                    return new VisitorStats(
                        "",
                        "",
                        commonJsonObj.getString("vc"),
                        commonJsonObj.getString("ch"),
                        commonJsonObj.getString("ar"),
                        commonJsonObj.getString("is_new"),
                        0L, 0L, 0L, 1L, 0L, jsonObj.getLong("ts")
                    );
                }
            }
        );

        //TODO 5.将上面3条流的数据 合并在一起
        DataStream<VisitorStats> unionDS = pageLogStatsDS
            .union(uvStatsDS, ujdStatsDS);

        //TODO 6.指定watermark以及提取事件时间字段
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithWatermarkDS = unionDS
            .assignTimestampsAndWatermarks(
                WatermarkStrategy.<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner(
                        new SerializableTimestampAssigner<VisitorStats>() {
                            @Override
                            public long extractTimestamp(VisitorStats visitorStats, long recordTimestamp) {
                                return visitorStats.getTs();
                            }
                        }
                    )
            );

        //TODO 7.分组  注意：如果按照mid进行分组，单位时间内，起不到很好的聚合效果；所以我们可以按照维度进行分组
        KeyedStream<VisitorStats, Tuple4<String, String, String, String>> visitorStatsKeyedDS = visitorStatsWithWatermarkDS.keyBy(
            new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
                @Override
                public Tuple4<String, String, String, String> getKey(VisitorStats visitorStats) throws Exception {
                    return Tuple4.of(
                        visitorStats.getVc(),
                        visitorStats.getAr(),
                        visitorStats.getCh(),
                        visitorStats.getIs_new()
                    );
                }
            }
        );
        //TODO 8.开窗
        WindowedStream<VisitorStats, Tuple4<String, String, String, String>, TimeWindow> windowDS = visitorStatsKeyedDS
            .window(TumblingEventTimeWindows.of(Time.seconds(10)));

        //TODO 9.聚合计算
        SingleOutputStreamOperator<VisitorStats> reduceDS = windowDS.reduce(
            new ReduceFunction<VisitorStats>() {
                @Override
                public VisitorStats reduce(VisitorStats stats1, VisitorStats stats2) throws Exception {
                    stats1.setPv_ct(stats1.getPv_ct() + stats2.getPv_ct());
                    stats1.setUv_ct(stats1.getUv_ct() + stats2.getUv_ct());
                    stats1.setUj_ct(stats1.getUj_ct() + stats2.getUj_ct());
                    stats1.setSv_ct(stats1.getSv_ct() + stats2.getSv_ct());
                    stats1.setDur_sum(stats1.getDur_sum() + stats2.getDur_sum());
                    return stats1;
                }
            },
            new ProcessWindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
                @Override
                public void process(Tuple4<String, String, String, String> tuple4, Context context, Iterable<VisitorStats> elements, Collector<VisitorStats> out) throws Exception {
                    for (VisitorStats visitorStats : elements) {
                        visitorStats.setStt(DateTimeUtil.toYmdhms(new Date(context.window().getStart())));
                        visitorStats.setEdt(DateTimeUtil.toYmdhms(new Date(context.window().getEnd())));
                        visitorStats.setTs(System.currentTimeMillis());
                        out.collect(visitorStats);
                    }
                }
            }
        );

        reduceDS.print(">>>>>>");
        //TODO 10.将聚合的结果写到OLAP数据库中（ClickHouse）
        reduceDS.addSink(
            ClickHouseUtil.getJdbcSink("insert into visitor_stats_0701 values(?,?,?,?,?,?,?,?,?,?,?,?)")
        );

        env.execute();
    }
}
